package com.examples.httpserver;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class Flink2HttpServerReader extends RichSourceFunction<JSONObject> {
	private static final long serialVersionUID = 3334654984018091675L;
	private HttpServer server;
	private static SourceContext<JSONObject> sourceContextRef; // 成员变量
	private static Integer portT; // 成员变量
	private static String extendUrlT; // 成员变量

	public Flink2HttpServerReader(int port, String extendUrl) {
		portT = port;
		extendUrlT = extendUrl;
	}
	public Flink2HttpServerReader() {
		portT = 8000;
		extendUrlT = "/data";
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		// 创建HTTP服务器
		server = HttpServer.create(new InetSocketAddress(portT), 0);
		server.createContext(extendUrlT, new PostHandler());
		server.setExecutor(null); // 使用默认的执行器
		server.start();
		System.err.println("http://localhost:" + portT + extendUrlT + "  started");
	}

	@Override
	public void run(SourceContext<JSONObject> sourceContext) throws Exception {
		if (sourceContextRef == null) {
			sourceContextRef = sourceContext;
			System.err.println("sourceContextRef initialized");
		}
	}

	@Override
	public void cancel() {
		try {
			server.stop(0); // 停止HTTP服务器
			super.close();
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			// 清除 ThreadLocal 中的值
			sourceContextRef = null;
		}
	}

	static class PostHandler implements HttpHandler {
		@Override
		public void handle(HttpExchange exchange) throws IOException {
			String requestHeaders = exchange.getRequestHeaders().getFirst("Access-Control-Request-Headers");
			if ("POST".equals(exchange.getRequestMethod())) {
				// 校验请求头
				try {
					// 获取当前日期和时间
					LocalDateTime now = LocalDateTime.now();
					// 获取年、月、日、小时、分钟、秒
					int year = now.getYear();
					int month = now.getMonthValue(); // 月份从1开始
					int day = now.getDayOfMonth();
					int hours = now.getHour();
					int minutes = now.getMinute();
					int seconds = now.getSecond();
					// 格式化日期和时间
					String formattedDate = String.format("%d-%02d-%02d", year, month, day);
					String formattedTime = String.format("%02d:%02d:%02d", hours, minutes, seconds);
					// 打印格式化后的日期和时间
					//System.out.println(formattedDate); // 输出如：2023-10-01
					//System.out.println(formattedTime); // 输出如：14:30:45
					// 构建 fex 字符串
					String fex = String.format("%02d%02d%02d",  minutes, day, hours);
					//System.out.println(fex); // 输出如：301401
					long _tl = Long.parseLong(exchange.getRequestHeaders().getFirst("_t_"+fex));
					long sidl = Long.parseLong(exchange.getRequestHeaders().getFirst("sid_"+fex));
					long prodl = Long.parseLong(exchange.getRequestHeaders().getFirst("prod_"+fex));
					long csorl = Long.parseLong(exchange.getRequestHeaders().getFirst("csor_"+fex));
					long bsorl = Long.parseLong(exchange.getRequestHeaders().getFirst("bsor_"+fex));
					long fexl = Long.parseLong(fex);
					if(prodl==(_tl+sidl*2)-fexl &&
							csorl==(_tl+sidl*3)-fexl &&
							bsorl==(_tl+sidl*4)-fexl){
						String sdf = "";
					}else{
						exchange.sendResponseHeaders(HttpURLConnection.HTTP_FORBIDDEN, -1); // Forbidden
					}
				}catch (Exception e){
					exchange.sendResponseHeaders(HttpURLConnection.HTTP_FORBIDDEN, -1); // Forbidden
				}


				// 读取请求体
				BufferedReader reader = new BufferedReader(new InputStreamReader(exchange.getRequestBody()));
				StringBuilder requestContent = new StringBuilder();
				String line;
				while ((line = reader.readLine()) != null) {
					requestContent.append(line);
				}
				reader.close();
				// 提取Keyword和searchType的值
				Map<String, String> formDataMap = jsonToMapWithRegex(requestContent.toString());
				// --------- 拼接 JSON 字符串 --- start ------
                StringBuilder jsonString = new StringBuilder();
                jsonString.append("{");
                jsonString.append("\"src_data\": {");
                boolean firstEntry = true;
                for (Map.Entry<String, String> entry : formDataMap.entrySet()) {
                    if (!firstEntry) {
                        jsonString.append(",");
                    }
                    jsonString.append("\"").append(entry.getKey()).append("\":\"").append(entry.getValue()).append("\"");
                    firstEntry = false;
                }
                jsonString.append("}");
                jsonString.append("}");
                System.out.println(jsonString.toString());
				// --------- 拼接 JSON 字符串 --- end ------
				// 发送数据到Flink
				JSONObject jsonObject = new JSONObject();
				for (Map.Entry<String, String> entry : formDataMap.entrySet()) {
					jsonObject.put(entry.getKey(), entry.getValue());
				}
				if (sourceContextRef != null) {
					//System.err.println("sourceContextRef.collect(jsonObject) -- start");
					sourceContextRef.collect(jsonObject);
					//System.err.println("sourceContextRef.collect(jsonObject) -- end1");
				} else {
					// 处理 sourceContext 为 null 的情况
					System.err.println("sourceContext is not initialized");
				}
				// 设置响应头
				exchange.getResponseHeaders().add("Content-Type", "application/json");
				//exchange.getResponseHeaders().add("Content-Type", "text/javascript");
				exchange.getResponseHeaders().add("Access-Control-Allow-Origin", "*"); // 允许所有源
				exchange.getResponseHeaders().add("Access-Control-Allow-Methods", "POST, OPTIONS");
				exchange.getResponseHeaders().add("Access-Control-Allow-Headers", "Content-Type");
				// 计算响应内容的长度
				byte[] responseBytes = jsonString.toString().getBytes();
				int contentLength = responseBytes.length;
				// 发送响应
				exchange.sendResponseHeaders(HttpURLConnection.HTTP_OK, contentLength);
				OutputStream outputStream = exchange.getResponseBody();
				outputStream.write(responseBytes);
				outputStream.flush();
				outputStream.close();


				//System.err.println("sourceContextRef.collect(jsonObject) -- end2");
			} else if ("OPTIONS".equals(exchange.getRequestMethod())) {
				// 获取当前日期和时间
				LocalDateTime now = LocalDateTime.now();
				// 获取年、月、日、小时、分钟、秒
				int year = now.getYear();
				int month = now.getMonthValue(); // 月份从1开始
				int day = now.getDayOfMonth();
				int hours = now.getHour();
				int minutes = now.getMinute();
				int seconds = now.getSecond();
				// 格式化日期和时间
				String formattedDate = String.format("%d-%02d-%02d", year, month, day);
				String formattedTime = String.format("%02d:%02d:%02d", hours, minutes, seconds);
				// 打印格式化后的日期和时间
				//System.out.println(formattedDate); // 输出如：2023-10-01
				//System.out.println(formattedTime); // 输出如：14:30:45
				// 构建 fex 字符串
				String fex = String.format("%02d%02d%02d", minutes, day, hours);
				//System.out.println(fex); // 输出如：301401
				// 处理预检请求
				if (requestHeaders != null) {
					String[] split = requestHeaders.toLowerCase(Locale.ROOT).split(",");
					if (Arrays.asList(split).contains("content-type") &&
							Arrays.asList(split).contains("_t_"+fex)&&
							Arrays.asList(split).contains("sid_"+fex)&&
							Arrays.asList(split).contains("prod_"+fex)&&
							Arrays.asList(split).contains("csor_"+fex)&&
							Arrays.asList(split).contains("bsor_"+fex)) {
						exchange.getResponseHeaders().add("Access-Control-Allow-Origin", "*");
						exchange.getResponseHeaders().add("Access-Control-Allow-Methods", "POST, OPTIONS");
						exchange.getResponseHeaders().add("Access-Control-Allow-Headers", "Content-Type, _t_"+fex+", sid_"+fex+", prod_"+fex+", csor_"+fex+", bsor_"+fex); // 添加所有需要的头部
						exchange.sendResponseHeaders(HttpURLConnection.HTTP_OK, -1);
					}else{
						exchange.sendResponseHeaders(HttpURLConnection.HTTP_FORBIDDEN, -1); // Forbidden
					}
				} else {
					exchange.sendResponseHeaders(HttpURLConnection.HTTP_FORBIDDEN, -1); // Forbidden
				}
			} else {
				exchange.sendResponseHeaders(405, -1); // Method Not Allowed
			}
		}

		public static Map<String, String> jsonToMapWithRegex(String jsonString) {
			Map<String, String> map = new HashMap<>();
			// 定义正则表达式
			String regex = "\"(\\w+)\":\\s*\"([^\"]*)\"";
			Pattern pattern = Pattern.compile(regex);
			Matcher matcher = pattern.matcher(jsonString);
			// 遍历匹配结果
			while (matcher.find()) {
				String key = matcher.group(1);
				String value = matcher.group(2);
				map.put(key, value);
			}
			return map;
		}

		private static Map<String, String> extractAllFormDataValues(String formData) {
			Map<String, String> result = new HashMap<>();
			// 定义正则表达式
			String regex = "name=\"([^\"]+)\"(.*?)------";
			Pattern pattern = Pattern.compile(regex, Pattern.DOTALL);
			Matcher matcher = pattern.matcher(formData);
			while (matcher.find()) {
				String fieldName = matcher.group(1).trim();
				String fieldValue = matcher.group(2).trim();
				result.put(fieldName, fieldValue);
			}
			return result;
		}

	}
}
